package cn.doitedu.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MD5Hash;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

/**
 * 数据操作： 增删改查
 */
public class DMLDemo {
    public static void main(String[] args) throws IOException {

        Configuration conf = HBaseConfiguration.create();
        Connection conn = ConnectionFactory.createConnection(conf);

        Table table = conn.getTable(TableName.valueOf("doit:students"));

        // 读数据源
        BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));
        String line = null;

        ArrayList<Put> puts = new ArrayList<>();

        long start = System.currentTimeMillis();
        while((line=br.readLine())!=null){
            String[] split = line.split(",");

            // 000001,zhangsan,18,male,beijing,18000,2018-07-01
            String rkMD5 = MD5Hash.getMD5AsHex(split[0].getBytes());

            // 构造一个用于封装KeyValue数据的put对象
            Put put = new Put(Bytes.toBytes(rkMD5));
            put.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("name"),Bytes.toBytes(split[1]));
            put.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("age"),Bytes.toBytes(Integer.parseInt(split[2])));
            put.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("gender"),Bytes.toBytes(split[3]));
            put.addColumn(Bytes.toBytes("extra_info"),Bytes.toBytes("city"),Bytes.toBytes(split[4]));
            put.addColumn(Bytes.toBytes("extra_info"),Bytes.toBytes("salary"),Bytes.toBytes(Integer.parseInt(split[5])));
            put.addColumn(Bytes.toBytes("extra_info"),Bytes.toBytes("graduate"),Bytes.toBytes(split[6]));

            // 把封装好的一行数据，先暂存在一个list中
            puts.add(put);

            // 判断list中的条数是否满足批次大小（10条一批）
            if(puts.size()>=10){
                // 插入数据
                table.put(puts);
                // 清空list
                puts.clear();
            }
        }

        // 最后一批
        if(puts.size()>0){
            table.put(puts);
        }

        long end = System.currentTimeMillis();


        // 关闭连接
        table.close();
        conn.close();

        System.out.println("总耗时： " + (end-start));

    }
}
